import tushare as ts
import datetime as dt
import pandas as pd
from datetime import datetime
from django.conf import settings
import logging,time,asyncio
from asgiref.sync import sync_to_async
from .models import *
from .constants import *
from django.db.models import Max
from django.db import transaction

logger = logging.getLogger(__name__)

# 创建异步的数据库操作函数
update_data_record = sync_to_async(DataUpdateRecord.update_record)

# 定义一个同步函数来获取所有股票列表（排除ST股票）
def get_stocks_list():
    return list(StockBasic.objects.all())

# 将同步函数转换为异步函数
get_all_stocks = sync_to_async(get_stocks_list)
def get_last_update_date(data_type):
    """
    获取指定数据类型的最后更新日期
    :param data_type: 数据类型，来自DATA_TYPES常量
    :return: 最后更新日期，如果没有记录则返回START_KDATA_DATE
    """
    
    if data_type == 'index_weekly':
        start_date =  START_KDATA_DATE_WEEK
    elif data_type == 'index_monthly':
        start_date = START_KDATA_DATE_MONTH
    else:
        start_date = START_KDATA_DATE
    try:
        record = DataUpdateRecord.objects.filter(data_type=data_type).order_by('-last_update_date').first()
        return record.last_update_date if record else start_date
    except Exception as e:
        logger.error(f"Error getting last update date for {data_type}: {str(e)}")
        return start_date

# 计算两个日期之间的交易日数量
def count_trading_days(start_date, end_date):
    """
    计算两个日期之间的交易日数量
    :param start_date: 开始日期，datetime.date对象或START_KDATA_DATE常量
    :param end_date: 结束日期，datetime.date对象
    :return: 交易日数量
    """
    if not start_date or not end_date:
        return 0
    
    # 如果start_date是START_KDATA_DATE常量，直接使用它（因为它已经是datetime.date对象）
    if start_date == START_KDATA_DATE:
        start_date = START_KDATA_DATE
    
    if start_date > end_date:#起始日期和结束日期相同时可能为一天
        return 0
    
    try:
        # 将日期转换为字符串格式YYYYMMDD
        start_str = start_date.strftime('%Y%m%d')
        end_str = end_date.strftime('%Y%m%d')
        
        # 查询交易日历
        trading_days = TradeDate.objects.filter(
            cal_date__gte=start_str,
            cal_date__lte=end_str,
            is_open=True
        ).count()
        
        return trading_days
    except Exception as e:
        logger.error(f"Error counting trading days: {str(e)}")
        return 0

# 将同步函数转换为异步函数
get_last_update_date_async = sync_to_async(get_last_update_date)
count_trading_days_async = sync_to_async(count_trading_days)

class TushareAPI_Emotion:
    def __init__(self, token=None):
        """
        初始化 Tushare API
        :param token: Tushare API token，如果未提供则从环境变量获取
        """
        self.token = token or getattr(settings, 'TUSHARE_TOKEN', None)
        if not self.token:
            raise ValueError("Tushare API token is required")
        
        # 初始化 Tushare API
        ts.set_token(self.token)
        self.pro = ts.pro_api()
        self.plate_to_create = []#当前更新的板块列表，作为全局变量用于更新板块的个股关系数据

    async def update_all_data(self, start_date=None, end_date=None, websocket=None):
        """
        更新所有数据
        :param start_date: 开始日期，可选
        :param end_date: 结束日期，可选
        :param websocket: WebSocket对象，可选
        """
        results = {}
        try:
            #但实际更新还是看数据库中存储的各类数据的更新日期
            logger.info(f"Starting update of all market data from {start_date} to {end_date}")
            # 转换日期为datetime.date对象，用于计算交易日
            start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date() if start_date else dt.date.today()
            end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date() if end_date else dt.date.today()
            
            #################################首先计算那些数据需要更新###############################################
            # 计算各数据类型的实际起始日期
            def get_actual_start_date(last_update_date):
                if not last_update_date:
                    return start_date_obj
                else:#以数据库中记录的各个表格时间为准
                    next_day = last_update_date + dt.timedelta(days=1)
                    return  next_day

            #1. 检查同花顺行业板块、概念板块及其关系数据是否需要更新（每日更新）
            ths_plate_last_update = await get_last_update_date_async(DATA_TYPES['THS_PLATE'])
            ths_plate_start_date = get_actual_start_date(ths_plate_last_update)
            ths_plate_trade_days = await count_trading_days_async(ths_plate_start_date, end_date_obj)
            
            #2. 检查同花顺指数行情数据是否需要更新（每日更新）
            ths_plate_quote_last_update = await get_last_update_date_async(DATA_TYPES['THS_PLATE_QUOTE'])
            ths_plate_quote_start_date = get_actual_start_date(ths_plate_quote_last_update)
            ths_plate_quote_trade_days = await count_trading_days_async(ths_plate_quote_start_date, end_date_obj)
            
            #3检查涨跌停列表数据是否需要更新（每日更新）
            limit_list_last_update = await get_last_update_date_async(DATA_TYPES['LIMIT_LIST'])
            limit_list_start_date = get_actual_start_date(limit_list_last_update)
            limit_list_trade_days = await count_trading_days_async(limit_list_start_date, end_date_obj)
            
            #4. 检查开盘啦题材数据是否需要更新（每日更新）
            block_kpl_last_update = await get_last_update_date_async(DATA_TYPES['BLOCK_KPL_ZT_DAILY'])
            block_kpl_start_date = get_actual_start_date(block_kpl_last_update)
            block_kpl_trade_days = await count_trading_days_async(block_kpl_start_date, end_date_obj)
            
            #5. 检查开盘啦题材个股成分数据是否需要更新（每日更新）
            block_kpl_stock_last_update = await get_last_update_date_async(DATA_TYPES['BLOCK_KPL_STOCK'])
            block_kpl_stock_start_date = get_actual_start_date(block_kpl_stock_last_update)
            block_kpl_stock_trade_days = await count_trading_days_async(block_kpl_stock_start_date, end_date_obj)
            
            #6. 检查开盘啦榜单数据是否需要更新（每日更新）
            block_kpl_list_last_update = await get_last_update_date_async(DATA_TYPES['BLOCK_KPL_LIST'])
            block_kpl_list_start_date = get_actual_start_date(block_kpl_list_last_update)
            block_kpl_list_trade_days = await count_trading_days_async(block_kpl_list_start_date, end_date_obj)

            #7. 检查龙虎榜每日统计单数据是否需要更新（每日更新）
            lhb_daily_last_update = await get_last_update_date_async(DATA_TYPES['MARKET_LHB_LIST'])
            lhb_daily_start_date = get_actual_start_date(lhb_daily_last_update)
            lhb_daily_trade_days = await count_trading_days_async(lhb_daily_start_date, end_date_obj)
            
            #8. 检查龙虎榜每日机构交易明细数据是否需要更新（每日更新）
            lhb_detail_last_update = await get_last_update_date_async(DATA_TYPES['MARKET_LHB_DETAIL'])
            lhb_detail_start_date = get_actual_start_date(lhb_detail_last_update)
            lhb_detail_trade_days = await count_trading_days_async(lhb_detail_start_date, end_date_obj)
            
            #9. 检查游资每日交易明细数据是否需要更新
            hm_daily_last_update = await get_last_update_date_async(DATA_TYPES['MARKET_HM_DETAIL'])
            hm_daily_start_date = get_actual_start_date(hm_daily_last_update)
            hm_daily_trade_days = await count_trading_days_async(hm_daily_start_date, end_date_obj)

            #10.检查市场同花顺和东方财富热榜数据是否需要更新
            market_hot_last_update = await get_last_update_date_async(DATA_TYPES['MARKET_HOT_LIST'])
            market_hot_start_date = get_actual_start_date(market_hot_last_update)
            market_hot_trade_days = await count_trading_days_async(market_hot_start_date, end_date_obj)
            
            #11.检查市场新闻数据是否需要更新


            # 设置总步骤数（包括同花顺相关数据和指数数据的更新）
            total_steps = (1 if ths_plate_trade_days > 0 else 0) \
                + (1 if ths_plate_quote_trade_days > 0 else 0) \
                + (1 if limit_list_trade_days > 0 else 0) \
                + (1 if block_kpl_trade_days > 0 else 0) \
                + (1 if block_kpl_stock_trade_days > 0 else 0) \
                + (1 if block_kpl_list_trade_days > 0 else 0) \
                + (1 if lhb_daily_trade_days > 0 else 0) \
                + (1 if lhb_detail_trade_days > 0 else 0) \
                + (1 if hm_daily_trade_days > 0 else 0) \
                + (1 if market_hot_trade_days > 0 else 0)
            # 初始化当前步骤数
            current_step = 0
            if websocket:
                await websocket.send_progress(
                    current=0,
                    total=total_steps,
                    message="开始下载数据..."
                )
            ########################################开始正式下载数据##############################################
            # 1.更新同花顺板块数据
            if websocket:
                current_step += 1
                await websocket.send_progress(
                    current=current_step,
                    total=total_steps,
                    message="正在更新同花顺板块数据..."
                )
            if ths_plate_trade_days > 0:
                results['ths_plate'] = await self.update_ths_plate()
                logger.info("Updated THS plate data")
            
                results['ths_stock_plate'] = await self.update_ths_stock_plate()
                logger.info("Updated THS stock industry relationship data")
            
            # 2.更新同花顺板块指数行情数据
            if ths_plate_quote_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新同花顺板块指数行情数据... (距上次更新: {ths_plate_quote_trade_days}个交易日)"
                    )
                
                # 使用批处理机制更新同花顺板块指数行情数据
                results['ths_plate_quote'] = await self.batch_update_ths_plate_quote(
                    start_date=ths_plate_quote_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated THS plate quote data for {ths_plate_quote_trade_days} trading days from {ths_plate_quote_start_date}")
            #3 涨跌停数据
            if limit_list_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新涨跌停数据... (距上次更新: {limit_list_trade_days}个交易日)"
                        )
                    results['limit_list'] = await self.update_limit_list(
                        start_date=limit_list_start_date.strftime('%Y%m%d'),
                        end_date=end_date_obj.strftime('%Y%m%d'),
                        websocket=websocket
                    )
                  
                logger.info(f"Updated limit list data for {limit_list_trade_days} trading days from {limit_list_start_date}")
            
            #4.更新开盘啦题材数据
            if block_kpl_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新开盘啦题材数据... (距上次更新: {block_kpl_trade_days}个交易日)"
                    )
                results['block_kpl_zt_daily'] = await self.update_block_kpl_zt_daily(
                    start_date=block_kpl_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated block kpl zt daily data for {block_kpl_trade_days} trading days from {block_kpl_start_date}")
            
            # #5.更新开盘啦个股成分数据
            if block_kpl_stock_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新开盘啦个股成分数据... (距上次更新: {block_kpl_stock_trade_days}个交易日)"
                    )
                results['block_kpl_stock'] = await self.update_block_kpl_stock(
                    start_date=block_kpl_stock_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated block kpl stock data for {block_kpl_stock_trade_days} trading days from {block_kpl_stock_start_date}")
            
            # #6.更新开盘啦榜单数据
            if block_kpl_list_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新开盘啦榜单数据... (距上次更新: {block_kpl_list_trade_days}个交易日)"
                    )
                results['block_kpl_list'] = await self.update_block_kpl_list(
                    start_date=block_kpl_list_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated block kpl list data for {block_kpl_list_trade_days} trading days from {block_kpl_list_start_date}")

            #7.更新龙虎榜每日统计单数据
            if lhb_daily_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新龙虎榜每日统计单数据... (距上次更新: {lhb_daily_trade_days}个交易日)"
                    )
                results['market_lhb_list'] = await self.update_market_lhb_list(
                    start_date=lhb_daily_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated market lhb list data for {lhb_daily_trade_days} trading days from {lhb_daily_start_date}")
            
            # #8.更新龙虎榜每日机构交易明细数据
            if lhb_detail_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新龙虎榜每日机构交易明细数据... (距上次更新: {lhb_detail_trade_days}个交易日)"
                    )
                results['market_lhb_detail'] = await self.update_market_lhb_detail(
                    start_date=lhb_detail_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
            #9.直接更新游资信息数据，不需要计算日期;更新游资交易信息，计算日期，且只更新最近两日（因为每天只有两次的权限）
            if hm_daily_trade_days > 0:
                results['market_hm_info'] = await self.update_market_hm_info()#先更新游资信息
                logger.info("Updated market hm info data")
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新游资信息数据... (距上次更新: {hm_daily_trade_days}个交易日)"
                    )
                results['market_hm_info'] = await self.update_market_hm_detail(
                    start_date=hm_daily_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated market hm info data for {hm_daily_trade_days} trading days from {hm_daily_start_date}")
            
            #10.更新同花顺和东方财富热榜数据
            if market_hot_trade_days > 0:
                if websocket:
                    current_step += 1
                    await websocket.send_progress(
                        current=current_step,
                        total=total_steps,
                        message=f"开始更新同花顺和东方财富热榜数据... (距上次更新: {market_hot_trade_days}个交易日)"
                    )
                results['market_hot'] = await self.update_market_hot(
                    start_date=market_hot_start_date.strftime('%Y%m%d'),
                    end_date=end_date_obj.strftime('%Y%m%d'),
                    websocket=websocket
                )
                logger.info(f"Updated market hot data for {market_hot_trade_days} trading days from {market_hot_start_date}")


            # 更新数据更新记录（无论是否有更新都记录）
            await update_data_record(
                DATA_TYPES['ALL_DATA_3'],
                end_date_obj
            )

            # 更新数据更新记录（无论是否有更新都记录）
            await update_data_record(
                DATA_TYPES['ALL_DATA'],
                end_date_obj
            )



        except Exception as e:
            logger.error(f"Error updating all market data: {str(e)}")
            raise
    async def update_ths_plate(self):
        """
        更新同花顺行业板块信息
        :return: 更新的记录数
        """
        total_count =0
        plate_type_list = ['I','N']
        for plate_type in plate_type_list:
            try:
                plate_name = '行业' if plate_type == 'I' else '概念'
                logger.info(f"开始更新同花顺{plate_name}板块信息")
                
                # 获取同花顺行业板块信息
                df = self.pro.ths_index(exchange='A', type=plate_type)
                df = df[df['ts_code'].str.startswith('88')]
                if df.empty:
                    logger.warning(f"未获取到同花顺{plate_name}板块数据")
                    return 0
                
                # 直接将DataFrame转换为模型实例列表
                @sync_to_async
                def bulk_create_ths_plate(df):
                    with transaction.atomic():
                        # 获取现有的ts_code列表
                        existing_ts_codes = set(ThsPlate.objects.values_list('ts_code', flat=True))
                        df['count'].fillna(0, inplace=True)#替换单个列的 `NaN` 值为 0
                        # 准备要创建和更新的记录
                        to_create = []
                        to_update = []
                        
                        for _, row in df.iterrows():
                            plate = ThsPlate(
                                ts_code=row['ts_code'],
                                name=row['name'],
                                count=row['count'],
                                exchange=row['exchange'],
                                list_date=row['list_date'] if 'list_date' in row else None,
                                type=row['type']
                            )
                            
                            if row['ts_code'] in existing_ts_codes:
                                to_update.append(plate)
                            else:
                                to_create.append(plate)
                                self.plate_to_create.append(row['ts_code'] )
                        
                        # 批量创建新记录
                        created_count = 0
                        if to_create:
                            ThsPlate.objects.bulk_create(to_create, batch_size=1000)
                            created_count = len(to_create)
                        
                        # 批量更新现有记录
                        updated_count = 0
                        if to_update:
                            # 使用bulk_update更新现有记录
                            ThsPlate.objects.bulk_update(
                                to_update, 
                                ['name', 'count', 'exchange', 'list_date', 'type'],
                                batch_size=1000
                            )
                            updated_count = len(to_update)
                        return created_count + updated_count
                
                # 执行批量创建和更新
                count = await bulk_create_ths_plate(df)
                logger.info(f"成功更新 {count} 条同花顺{plate_name}板块记录")
                total_count += count    
            except Exception as e:
                logger.error(f"更新同花顺{plate_name}板块数据时出错: {str(e)}")
                raise
        # 更新数据更新记录
        await update_data_record(
            DATA_TYPES['THS_PLATE'],
            dt.date.today()
        )  
        logger.info(f"成功更新 {total_count} 条同花顺行业与概念板块记录")
        return total_count
    
    async def update_ths_stock_plate(self):
        """
        更新股票所属行业板块关系
        :return: 更新的记录数
        """
        call_count = 0
        start_time = time.time()
        total_time = 0
        total_count = 0#行业和概念2个板块的更新个数
        try:
            logger.info(f"开始更新股票所属板块关系")
            # 获取所有板块代码
            @sync_to_async
            def get_plate_codes():
                return list(ThsPlate.objects.values_list('ts_code', flat=True))
            
            #判断是否超过20个交易日。如果超过则全部更新所有板块，如果不超过，则只更新当天新增的板块
            stock_plate_last_update = await get_last_update_date_async(DATA_TYPES['THS_STOCK_PLATE'])
            stock_plate_trade_days = await count_trading_days_async(stock_plate_last_update, dt.date.today())
            if stock_plate_trade_days > PLATE_REFRESH_DAYS_INTERVAL:
                plate_codes = await get_plate_codes()
                if not plate_codes:
                    logger.warning(f"未找到同花顺板块数据，请先更新板块信息")
                    return 0
            elif len(self.plate_to_create) > 0:
                plate_codes = self.plate_to_create
            else:
                plate_codes = []
                return 0
            # 对每个板块获取成分股（包括了行业和概念板块）
            for plate_code in plate_codes:
                try:
                    # 获取行业板块成分股
                    call_count += 1
                    if call_count == TUSHARE_API_LIMIT['ths_member_max_in_minute']:
                        total_time = time.time() - start_time
                        print(f'运行500次访问总共花费时间{total_time}秒')
                        if  total_time < SECONDS_IN_MINUTE:
                            await asyncio.sleep(SECONDS_IN_MINUTE - total_time)
                    df = self.pro.ths_member(ts_code=plate_code,fields=['ts_code','con_code','con_name','is_new'])
                    if df.empty:
                        continue
                    # 直接将DataFrame转换为模型实例列表
                    @sync_to_async
                    def bulk_create_stock_plate(df, plate_code):
                         # 获取需要删除的记录。如果该板块需要更新，则删除历史记录，直接建新记录
                        existing_records = ThsStockPlate.objects.filter(ts_code=plate_code)
                        # 删除已有记录
                        existing_records.delete()
                        with transaction.atomic():
                            # 准备要创建和更新的记录
                            to_create = []
                            
                            for _, row in df.iterrows():
                                stock_plate = ThsStockPlate(
                                    ts_code=plate_code,
                                    con_code=row['con_code'],
                                    con_name=row['con_name'],
                                    is_new=row['is_new'] == 'Y'
                                )
                                # 创建新记录
                                to_create.append(stock_plate)
                            
                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                ThsStockPlate.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)
                            return created_count
                    
                    # 执行批量创建和更新
                    count = await bulk_create_stock_plate(df, plate_code)
                    total_count += count
                    
                except Exception as e:
                    logger.error(f"更新板块 {plate_code} 成分股时出错: {str(e)}")
                    continue
        except Exception as e:
            logger.error(f"更新股票所属板块关系时出错: {str(e)}")
            raise
        # 更新数据更新记录
        await update_data_record(
            DATA_TYPES['THS_STOCK_PLATE'],
            dt.date.today()
        )    
        logger.info(f"成功更新 {total_count} 条股票行业与概念板块关系记录")
        return total_count    
    async def update_ths_stock_plate(self):
        """
        更新股票所属行业板块关系
        :return: 更新的记录数
        """
        call_count = 0
        start_time = time.time()
        total_time = 0
        total_count = 0#行业和概念2个板块的更新个数
        try:
            logger.info(f"开始更新股票所属板块关系")
            # 获取所有板块代码
            @sync_to_async
            def get_plate_codes():
                return list(ThsPlate.objects.values_list('ts_code', flat=True))
            
            #判断是否超过20个交易日。如果超过则全部更新所有板块，如果不超过，则只更新当天新增的板块
            stock_plate_last_update = await get_last_update_date_async(DATA_TYPES['THS_STOCK_PLATE'])
            stock_plate_trade_days = await count_trading_days_async(stock_plate_last_update, dt.date.today())
            if stock_plate_trade_days > PLATE_REFRESH_DAYS_INTERVAL:
                plate_codes = await get_plate_codes()
                if not plate_codes:
                    logger.warning(f"未找到同花顺板块数据，请先更新板块信息")
                    return 0
            elif len(self.plate_to_create) > 0:
                plate_codes = self.plate_to_create
            else:
                plate_codes = []
                return 0
            # 对每个板块获取成分股（包括了行业和概念板块）
            for plate_code in plate_codes:
                try:
                    # 获取行业板块成分股
                    call_count += 1
                    if call_count == TUSHARE_API_LIMIT['ths_member_max_in_minute']:
                        total_time = time.time() - start_time
                        print(f'运行500次访问总共花费时间{total_time}秒')
                        if  total_time < SECONDS_IN_MINUTE:
                            await asyncio.sleep(SECONDS_IN_MINUTE - total_time)
                    df = self.pro.ths_member(ts_code=plate_code,fields=['ts_code','con_code','con_name','is_new'])
                    if df.empty:
                        continue
                    # 直接将DataFrame转换为模型实例列表
                    @sync_to_async
                    def bulk_create_stock_plate(df, plate_code):
                         # 获取需要删除的记录。如果该板块需要更新，则删除历史记录，直接建新记录
                        existing_records = ThsStockPlate.objects.filter(ts_code=plate_code)
                        # 删除已有记录
                        existing_records.delete()
                        with transaction.atomic():
                            # 准备要创建和更新的记录
                            to_create = []
                            
                            for _, row in df.iterrows():
                                stock_plate = ThsStockPlate(
                                    ts_code=plate_code,
                                    con_code=row['con_code'],
                                    con_name=row['con_name'],
                                    is_new=row['is_new'] == 'Y'
                                )
                                # 创建新记录
                                to_create.append(stock_plate)
                            
                            # 批量创建新记录
                            created_count = 0
                            if to_create:
                                ThsStockPlate.objects.bulk_create(to_create, batch_size=1000)
                                created_count = len(to_create)
                            return created_count
                    
                    # 执行批量创建和更新
                    count = await bulk_create_stock_plate(df, plate_code)
                    total_count += count
                    
                except Exception as e:
                    logger.error(f"更新板块 {plate_code} 成分股时出错: {str(e)}")
                    continue
        except Exception as e:
            logger.error(f"更新股票所属板块关系时出错: {str(e)}")
            raise
        # 更新数据更新记录
        await update_data_record(
            DATA_TYPES['THS_STOCK_PLATE'],
            dt.date.today()
        )    
        logger.info(f"成功更新 {total_count} 条股票行业与概念板块关系记录")
        return total_count    
    
    async def batch_update_ths_plate_quote(self, start_date=None, end_date=None, websocket=None):
        """
        批量更新所有同花顺板块的指数行情数据
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :param websocket: 可选的WebSocket连接，用于发送进度更新
        :return: 更新的记录数
        """
        logger.info(f'Starting batch update of all ths plate quote data from {start_date} to {end_date}')
        total_count = 0
        try:
            # 获取所有同花顺板块列表
            @sync_to_async
            def get_all_plates():
                return list(ThsPlate.objects.all())
            
            plates = await get_all_plates()
            
            # 解析日期
            try:
                if start_date and end_date:
                    if '-' in start_date:  # 格式为YYYY-MM-DD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                    
                    if '-' in end_date:  # 格式为YYYY-MM-DD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                    else:  # 格式为YYYYMMDD
                        end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
                else:
                    # 如果没有指定日期范围，默认更新最近10个交易日
                    end_date_obj = dt.date.today()
                    start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日
            except Exception as e:
                logger.error(f"Error parsing dates: {str(e)}")
                raise
            
            # 获取Tushare API限制
            max_data_count = TUSHARE_API_LIMIT.get('ths_plate_quote_max_count', 3000)  # 每次调用最多返回的数据条数
            max_plate_num = TUSHARE_API_LIMIT.get('ths_plate_quote_max_num', 100)      # 每次调用最多的板块数量
            
            # 计算每批次板块可以获取的最大交易日数
            max_days_per_batch = max_data_count // max_plate_num  # 通常是30天
            
            # 获取所有交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))
            
            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0
            
            # 将板块列表分成多个批次
            plate_batches = [plates[i:i + max_plate_num] for i in range(0, len(plates), max_plate_num)]
            total_batches = len(plate_batches)
            
            # 将交易日列表分成多个时间段
            date_batches = [trade_dates[i:i + max_days_per_batch] for i in range(0, len(trade_dates), max_days_per_batch)]
            total_date_batches = len(date_batches)
            
            batch_count = 0
            total_batch_count = total_batches * total_date_batches
            
            # 对每个板块批次和每个时间段进行处理
            for plate_batch_index, plate_batch in enumerate(plate_batches, 1):
                ts_codes = [plate.ts_code for plate in plate_batch]
                
                for date_batch_index, date_batch in enumerate(date_batches, 1):
                    batch_count += 1
                    
                    # 发送进度更新
                    if websocket:
                        message = (f"正在更新第 {plate_batch_index}/{total_batches} 批次, "
                                 f"第 {date_batch_index}/{total_date_batches} 个时间段 "
                                 f"({len(date_batch)}个交易日, {len(plate_batch)}个板块指数行情数据)")
                        try:
                            await websocket.send_progress(
                                current=batch_count,
                                total=total_batch_count,
                                message=message
                            )
                        except Exception as e:
                            logger.error(f"WebSocket发送失败: {str(e)}")
                    
                    try:
                        # 批量更新数据
                        count = await self.update_batch_ths_plate_quote(
                            ts_codes=ts_codes,
                            start_date=date_batch[0],
                            end_date=date_batch[-1]
                        )
                        total_count += count
                    except Exception as e:
                        logger.error(f"Error updating batch {batch_count}/{total_batch_count}: {str(e)}")
                        continue
            
            # 更新数据更新记录
            if end_date:
                # 根据end_date的格式进行不同的处理
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                end_date_obj = dt.date.today()
                
            await update_data_record(
                DATA_TYPES['THS_PLATE_QUOTE'],
                end_date_obj
            )
            
            logger.info(f"Successfully updated {total_count} ths plate quote records in total")
            return total_count
            
        except Exception as e:
            logger.error(f"Error in batch update: {str(e)}")
            raise
     #更新板块与个股关系数据时，不需要区分概念还是行业，因为他们都存在一个表里。
    async def update_batch_ths_plate_quote(self, ts_codes, start_date=None, end_date=None):
        """
        批量更新多个同花顺板块的指数行情数据
        :param ts_codes: 板块代码列表（Tushare格式）
        :param start_date: 开始日期，格式：YYYYMMDD
        :param end_date: 结束日期，格式：YYYYMMDD
        :return: 更新的记录数
        """
        try:
            # 将板块代码列表转换为字符串，用逗号分隔
            ts_codes_str = ','.join(ts_codes)
            
            # 获取同花顺指数行情数据
            df = self.pro.ths_daily(
                ts_code=ts_codes_str,
                start_date=start_date,
                end_date=end_date,
                fields=["ts_code","trade_date","open","high","low","close","pre_close","change","pct_change","vol","turnover_rate","float_mv"]
            )
            
            if df.empty:
                return 0
            
            # 直接将DataFrame转换为模型实例列表
            @sync_to_async
            def bulk_create_ths_plate_quote(df):
                with transaction.atomic():
                    # 创建唯一键列表，用于避免重复数据
                    unique_keys = set()
                    for _, row in df.iterrows():
                        unique_keys.add((row['ts_code'], row['trade_date']))
                    
                    # 获取现有的记录.先保留，如果后续有效率问题，可以考虑删除这个步骤。
                    existing_records = {
                        (record.ts_code, record.trade_date): record 
                        for record in ThsPlateQuote.objects.filter(
                            ts_code__in=df['ts_code'].unique(),
                            trade_date__gte=start_date,
                            trade_date__lte=end_date
                        )
                    }
                    
                    # 准备要创建和更新的记录
                    to_create = []
                    to_update = []
                    
                    for _, row in df.iterrows():
                        record_key = (row['ts_code'], row['trade_date'])
                        
                        # 避免处理重复数据
                        if record_key in unique_keys:
                            unique_keys.remove(record_key)  # 移除已处理的键
                            
                            plate_quote = ThsPlateQuote(
                                ts_code=row['ts_code'],
                                trade_date=row['trade_date'],
                                open=row['open'],
                                high=row['high'],
                                low=row['low'],
                                close=row['close'],
                                pre_close=row['pre_close'],
                                change=row['change'],
                                pct_chg=row['pct_change'],
                                vol=row['vol'],
                                turnover_rate=row['turnover_rate'],
                                float_mv=row['float_mv']
                            )
                            
                            if record_key in existing_records:
                                # 更新现有记录
                                existing_record = existing_records[record_key]
                                existing_record.open = row['open']
                                existing_record.high = row['high']
                                existing_record.low = row['low']
                                existing_record.close = row['close']
                                existing_record.pre_close = row['pre_close']
                                existing_record.change = row['change']
                                existing_record.pct_chg = row['pct_change']
                                existing_record.vol = row['vol']
                                existing_record.turnover_rate = row['turnover_rate']
                                existing_record.float_mv = row['float_mv']
                                to_update.append(existing_record)
                            else:
                                # 创建新记录
                                to_create.append(plate_quote)
                    
                    # 批量创建新记录
                    batch_size = TUSHARE_API_LIMIT.get('ths_plate_quote_max_count', 3000)  # 每次调用最多返回的数据条数
                    created_count = 0
                    if to_create:
                        ThsPlateQuote.objects.bulk_create(to_create, batch_size=batch_size)
                        created_count = len(to_create)
                    
                    # 批量更新现有记录
                    updated_count = 0
                    if to_update:
                        ThsPlateQuote.objects.bulk_update(
                            to_update, 
                            ['open', 'high', 'low', 'close', 'pre_close', 'change', 'pct_chg', 'vol', 'turnover_rate', 'float_mv'],
                            batch_size=batch_size
                        )
                        updated_count = len(to_update)
                    
                    return created_count + updated_count
            
            # 执行批量创建和更新
            count = await bulk_create_ths_plate_quote(df)
            return count
            
        except Exception as e:
            logger.error(f"Error updating batch ths plate quote data: {str(e)}")
            raise
    async def update_limit_list(self, start_date=None,end_date=None,websocket=None):
        """
        更新涨跌停列表数据
        :param trade_date: 交易日期，格式：YYYYMMDD，如果未提供则使用当天日期
        :return: 更新的记录数
        """
        # 解析日期
        try:
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近1年的数据
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=365)  # 约1年
        except Exception as e:
            logger.error(f"Error parsing dates: {str(e)}")
            raise
        # 获取所有交易日列表
        trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
            cal_date__gte=start_date_obj.strftime('%Y%m%d'),
            cal_date__lte=end_date_obj.strftime('%Y%m%d'),
            is_open=True
        ).order_by('cal_date').values_list('cal_date', flat=True))
        if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

        total_dates = len(trade_dates)
        print(f'total_dates = {total_dates}')
        count_all = 0
        call_count = 0
        
        try:
            for i,trade_date in enumerate(trade_dates):
                if call_count == 0:#表示重新一轮计时
                    start_time = time.time()#一次下载数据如果天数很多的话，可能需要等待多次，就需要在每次等待完成后重新开始计时，通过call_count来控制
                # 发送进度更新
                if websocket:
                    message = f"正在更新每日涨跌停数据 trade_date = {trade_date}"
                    try:
                        await websocket.send_progress(
                            current=i,
                            total=total_dates,
                            message=message
                        )
                    except Exception as e:
                        logger.error(f"WebSocket发送失败: {str(e)}")
        
                print(f'trade_date = {trade_date}')
                # 获取涨跌停列表数据
                df = self.pro.limit_list_d(
                    trade_date=trade_date,
                    fields='trade_date,ts_code,industry,name,close,pct_chg,amount,limit_amount,float_mv, \
                                turnover_ratio,fd_amount,first_time,last_time,open_times,up_stat,limit_times,limit'
                )
                # print(f'df = {df}')
                call_count += 1
                if call_count == TUSHARE_API_LIMIT['limit_list_d_max_in_minute']:
                    total_time = time.time() - start_time
                    print(f'运行200次访问总共花费时间{total_time}秒')
                    if  total_time < SECONDS_IN_MINUTE:
                        await asyncio.sleep(SECONDS_IN_MINUTE - total_time)  
                        call_count = 0#需要清零，这样第二次还需要等待
                df.fillna(0, inplace=True)
                if df.empty:
                    logger.info(f"No limit list data found for date {trade_date}")
                    return 0
                
                # 直接将DataFrame转换为模型实例列表
                @sync_to_async
                def bulk_create_limit_list(df):
                    with transaction.atomic():
                        # 获取现有的记录
                        existing_records = {
                            (record.ts_code, record.trade_date): record 
                            for record in LimitList.objects.filter(
                                trade_date=trade_date
                            )
                        }
                        
                        # 准备要创建和更新的记录
                        to_create = []
                        to_update = []
                        
                        for _, row in df.iterrows():
                            record_key = (row['ts_code'], row['trade_date'])
                            
                            limit_data = LimitList(
                                trade_date=row['trade_date'],
                                ts_code=row['ts_code'],
                                industry=row['industry'],
                                name=row['name'],
                                close=row['close'],
                                pct_chg=row['pct_chg'],
                                amount=row['amount'],
                                limit_amount=row['limit_amount'],
                                float_mv=row['float_mv'],
                                turnover_ratio=row['turnover_ratio'],
                                fd_amount=row['fd_amount'],
                                first_time=row['first_time'],
                                last_time=row['last_time'],
                                open_times=row['open_times'],
                                up_stat=row['up_stat'],
                                limit_times=row['limit_times'],
                                limit=row['limit']
                            )
                            
                            if record_key in existing_records:
                                # 更新现有记录
                                existing_record = existing_records[record_key]
                                for field in ['industry','name', 'close', 'pct_chg', 'amount', 'limit_amount', 'float_mv', 'turnover_ratio',
                                            'fd_amount', 'first_time', 'last_time', 'open_times', 'up_stat', 'limit_times','limit']:
                                    setattr(existing_record, field, getattr(limit_data, field))
                                to_update.append(existing_record)
                            else:
                                # 创建新记录
                                to_create.append(limit_data)
                        
                        # 批量创建新记录
                        created_count = 0
                        if to_create:
                            LimitList.objects.bulk_create(to_create, batch_size=1000)
                            created_count = len(to_create)
                        
                        # 批量更新现有记录
                        updated_count = 0
                        if to_update:
                            LimitList.objects.bulk_update(
                                to_update, 
                                ['close', 'pct_chg', 'amount', 'limit_amount', 'float_mv', 'turnover_ratio',
                                    'fd_amount', 'first_time', 'last_time', 'open_times', 'up_stat', 'limit_times','limit'],
                                batch_size=1000
                            )
                            updated_count = len(to_update)
                        
                        return created_count + updated_count
                
                # 执行批量创建和更新
                count = await bulk_create_limit_list(df)
                count_all += count
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['LIMIT_LIST'],
                dt.datetime.strptime(trade_dates[-1], '%Y%m%d').date()
            )
            
            logger.info(f"Successfully updated {count} limit list records for date {trade_date}")
            return count_all
        except Exception as e:
            logger.error(f"Error updating limit list data: {str(e)}")
            raise
    async def update_block_kpl_zt_daily(self, start_date=None, end_date=None, websocket=None):
        """
        更新开盘啦题材数据
        :param start_date: 开始日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param end_date: 结束日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        # 解析日期
        try:
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近1年的数据
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=365)  # 约1年
        except Exception as e:
            logger.error(f"Error parsing dates: {str(e)}")
            raise

        start_date_obj = max(start_date_obj, BLOCK_KPL_DATE_START)  # 确保开始日期不早于2024年11月1日

        # 获取所有交易日列表
        trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
            cal_date__gte=start_date_obj.strftime('%Y%m%d'),
            cal_date__lte=end_date_obj.strftime('%Y%m%d'),
            is_open=True
        ).order_by('cal_date').values_list('cal_date', flat=True))
        if not trade_dates:
            logger.warning("No trading days found in the specified date range")
            return 0

        total_dates = len(trade_dates)
        logger.info(f"Total trading days to process: {total_dates}")
        count_all = 0
        batch_size = 10  # 每10个交易日的数据合并存储一次
        batch_dfs = []  # 存储当前批次的数据

        try:
            for i, trade_date in enumerate(trade_dates):
                # 发送进度更新
                if websocket:
                    message = f"正在更新开盘啦题材数据 trade_date = {trade_date}"
                    try:
                        await websocket.send_progress(
                            current=i,
                            total=total_dates,
                            message=message
                        )
                    except Exception as e:
                        logger.error(f"WebSocket发送失败: {str(e)}")

                # 获取开盘啦题材数据
                df = self.pro.kpl_concept(trade_date=trade_date)
                if df.empty:
                    logger.warning(f"No data found for trade_date: {trade_date}")
                    continue

                # 检查数据库中是否已存在该日期的数据
                @sync_to_async
                def check_existing_data(trade_date):
                    return BlockConceptKPL.objects.filter(trade_date=trade_date).exists()
                
                exists = await check_existing_data(trade_date)
                if exists:
                    logger.info(f"Data for trade_date {trade_date} already exists, skipping")
                    continue

                batch_dfs.append(df)

                # 每10个交易日的数据合并存储一次
                if len(batch_dfs) >= batch_size or i == total_dates - 1:
                    combined_df = pd.concat(batch_dfs, ignore_index=True)
                    
                    # 存储数据
                    @sync_to_async
                    def save_batch_data(df):
                        with transaction.atomic():
                            # 批量创建新记录
                            records = [
                                BlockConceptKPL(**row)
                                for row in df.to_dict('records')
                            ]
                            BlockConceptKPL.objects.bulk_create(records, batch_size=1000)
                            return len(records)
                    
                    count = await save_batch_data(combined_df)
                    count_all += count
                    batch_dfs = []  # 清空当前批次
                    logger.info(f"Saved batch of {count} records")

        except Exception as e:
            logger.error(f"Error updating block kpl zt daily data: {str(e)}")
            raise

        # 更新数据更新记录
        await update_data_record(
            DATA_TYPES['BLOCK_KPL_ZT_DAILY'],
            end_date_obj
        )
        logger.info(f"Successfully updated {count_all} records for block kpl zt daily data")
        return count_all
    
    async def update_block_kpl_stock(self, start_date=None, end_date=None, websocket=None):
        """
        更新开盘啦个股成分数据
        :param start_date: 开始日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param end_date: 结束日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        # 解析日期
        try:
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近1年的数据
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=365)  # 约1年
        except Exception as e:
            logger.error(f"Error parsing dates: {str(e)}")
            raise
        start_date_obj = max(start_date_obj, BLOCK_KPL_DATE_START)  # 确保开始日期不早于2024年11月1日
        # 获取所有交易日列表
        trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
            cal_date__gte=start_date_obj.strftime('%Y%m%d'),
            cal_date__lte=end_date_obj.strftime('%Y%m%d'),
            is_open=True
        ).order_by('cal_date').values_list('cal_date', flat=True))
        if not trade_dates:
            logger.warning("No trading days found in the specified date range")
            return 0

        total_dates = len(trade_dates)
        logger.info(f"Total trading days to process: {total_dates}")
        count_all = 0
        batch_size = 10  # 每10个交易日的数据合并存储一次
        batch_dfs = []  # 存储当前批次的数据

        try:
            for i, trade_date in enumerate(trade_dates):
                # 发送进度更新
                if websocket:
                    message = f"正在更新开盘啦个股成分数据 trade_date = {trade_date}"
                    try:
                        await websocket.send_progress(
                            current=i,
                            total=total_dates,
                            message=message
                        )
                    except Exception as e:
                        logger.error(f"WebSocket发送失败: {str(e)}")

                # 获取开盘啦个股成分数据
                df = self.pro.kpl_concept_cons(trade_date=trade_date)
                if df.empty:
                    logger.warning(f"No data found for trade_date: {trade_date}")
                    continue

                # 检查数据库中是否已存在该日期的数据
                @sync_to_async
                def check_existing_data(trade_date):
                    return BlockConceptStockKpl.objects.filter(trade_date=trade_date).exists()
                
                exists = await check_existing_data(trade_date)
                if exists:
                    logger.info(f"Data for trade_date {trade_date} already exists, skipping")
                    continue

                batch_dfs.append(df)

                # 每10个交易日的数据合并存储一次
                if len(batch_dfs) >= batch_size or i == total_dates - 1:
                    combined_df = pd.concat(batch_dfs, ignore_index=True)
                    
                    # 存储数据
                    @sync_to_async
                    def save_batch_data(df):
                        with transaction.atomic():
                            # 批量创建新记录
                            records = [
                                BlockConceptStockKpl(**row)
                                for row in df.to_dict('records')
                            ]
                            BlockConceptStockKpl.objects.bulk_create(records, batch_size=1000)
                            return len(records)
                    
                    count = await save_batch_data(combined_df)
                    count_all += count
                    batch_dfs = []  # 清空当前批次
                    logger.info(f"Saved batch of {count} records")

        except Exception as e:
            logger.error(f"Error updating block kpl stock data: {str(e)}")
            raise

        # 更新数据更新记录
        await update_data_record(
            DATA_TYPES['BLOCK_KPL_STOCK'],
            end_date_obj
        )
        logger.info(f"Successfully updated {count_all} records for block kpl stock data")
        return count_all
    
    async def update_block_kpl_list(self, start_date=None, end_date=None, websocket=None):
        """
        更新开盘啦榜单数据
        :param start_date: 开始日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param end_date: 结束日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        # 解析日期
        try:
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近1年的数据
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=365)  # 约1年
        except Exception as e:
            logger.error(f"Error parsing dates: {str(e)}")
            raise
        start_date_obj = max(start_date_obj, BLOCK_KPL_DATE_START)  # 确保开始日期不早于2024年11月1日
        # 获取所有交易日列表
        trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
            cal_date__gte=start_date_obj.strftime('%Y%m%d'),
            cal_date__lte=end_date_obj.strftime('%Y%m%d'),
            is_open=True
        ).order_by('cal_date').values_list('cal_date', flat=True))
        if not trade_dates:
            logger.warning("No trading days found in the specified date range")
            return 0

        total_dates = len(trade_dates)
        logger.info(f"Total trading days to process: {total_dates}")
        count_all = 0
        batch_size = 10  # 每10个交易日的数据合并存储一次
        batch_dfs = []  # 存储当前批次的数据

        try:
            for i, trade_date in enumerate(trade_dates):
                # 发送进度更新
                if websocket:
                    message = f"正在更新开盘啦榜单数据 trade_date = {trade_date}"
                    try:
                        await websocket.send_progress(
                            current=i,
                            total=total_dates,
                            message=message
                        )
                    except Exception as e:
                        logger.error(f"WebSocket发送失败: {str(e)}")

                # 获取开盘啦榜单数据
                df = self.pro.kpl_list(trade_date=trade_date)
                if df.empty:
                    logger.warning(f"No data found for trade_date: {trade_date}")
                    continue

                # 检查数据库中是否已存在相同 ts_code 和 trade_date 的数据
                @sync_to_async
                def filter_existing_data(df):
                    existing_records = BlockConceptListKpl.objects.filter(
                        ts_code__in=df['ts_code'].tolist(),
                        trade_date__in=df['trade_date'].tolist()
                    ).values_list('ts_code', 'trade_date')
                    existing_pairs = set((record[0], record[1]) for record in existing_records)
                    return df[~df.apply(lambda row: (row['ts_code'], row['trade_date']) in existing_pairs, axis=1)]
                
                filtered_df = await filter_existing_data(df)
                if filtered_df.empty:
                    logger.info(f"All data for trade_date {trade_date} already exists, skipping")
                    continue

                batch_dfs.append(filtered_df)

                # 每10个交易日的数据合并存储一次
                if len(batch_dfs) >= batch_size or i == total_dates - 1:
                    combined_df = pd.concat(batch_dfs, ignore_index=True)
                    
                    # 存储数据
                    @sync_to_async
                    def save_batch_data(df):
                        with transaction.atomic():
                            # 批量创建新记录
                            records = [
                                BlockConceptListKpl(**row)
                                for row in df.to_dict('records')
                            ]
                            try:
                                BlockConceptListKpl.objects.bulk_create(records, batch_size=1000, ignore_conflicts=True)
                                return len(records)
                            except Exception as e:
                                logger.warning(f"部分数据已存在，跳过重复数据: {str(e)}")
                                return 0
                    
                    count = await save_batch_data(combined_df)
                    count_all += count
                    batch_dfs = []  # 清空当前批次
                    logger.info(f"Saved batch of {count} records")

        except Exception as e:
            logger.error(f"Error updating block kpl list data: {str(e)}")
            raise

        # 更新数据更新记录
        await update_data_record(
            DATA_TYPES['BLOCK_KPL_LIST'],
            end_date_obj
        )
        logger.info(f"Successfully updated {count_all} records for block kpl list data")
        return count_all
    
    async def update_market_lhb_list(self, start_date=None, end_date=None, websocket=None):
        """
        更新龙虎榜每日统计单数据
        :param start_date: 开始日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param end_date: 结束日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        logger.info("Updating market lhb list data...start_date = %s, end_date = %s", start_date, end_date)
        # 解析日期
        try:
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近1年的数据
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=365)
        except Exception as e:
            logger.error(f"Error parsing dates: {str(e)}")
            raise

        # 获取所有交易日列表
        trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
            cal_date__gte=start_date_obj.strftime('%Y%m%d'),
            cal_date__lte=end_date_obj.strftime('%Y%m%d'),
            is_open=True
        ).order_by('cal_date').values_list('cal_date', flat=True))
        if not trade_dates:
            logger.warning("No trading days found in the specified date range")
            return 0

        total_dates = len(trade_dates)
        logger.info(f"Total trading days to process: {total_dates}")
        count_all = 0
        batch_size = 10  # 每10个交易日的数据合并存储一次
        batch_dfs = []  # 存储当前批次的数据
        call_count = 0
        start_time = time.time()

        try:
            for i, trade_date in enumerate(trade_dates):
                # logger.info(f"Updating market lhb list data for trade_date = {trade_date}")
                # 发送进度更新
                if websocket:
                    message = f"正在更新龙虎榜每日统计单数据 trade_date = {trade_date}"
                    try:
                        await websocket.send_progress(
                            current=i,
                            total=total_dates,
                            message=message
                        )
                    except Exception as e:
                        logger.error(f"WebSocket发送失败: {str(e)}")

                # 检查调用频率
                call_count += 1
                if call_count >= TUSHARE_API_LIMIT['market_lhb_top_list_max_in_minute']:
                    elapsed_time = time.time() - start_time
                    if elapsed_time < SECONDS_IN_MINUTE:  # 如果不足1分钟
                        await asyncio.sleep(SECONDS_IN_MINUTE - elapsed_time)
                    # 重置计数器和时间记录
                    call_count = 0
                    start_time = time.time()
                
                # 获取龙虎榜每日统计单数据
                df = self.pro.top_list(trade_date=trade_date)
                if df.empty:
                    logger.warning(f"No data found for trade_date: {trade_date}")
                    continue

                # 检查数据库中是否已存在相同 trade_date 和 ts_code 的数据
                @sync_to_async
                def filter_existing_data(df):
                    existing_records = MarketTopList.objects.filter(
                        trade_date__in=df['trade_date'].tolist(),
                        ts_code__in=df['ts_code'].tolist()
                    ).values_list('trade_date', 'ts_code')
                    existing_pairs = set((record[0], record[1]) for record in existing_records)
                    return df[~df.apply(lambda row: (row['trade_date'], row['ts_code']) in existing_pairs, axis=1)]
                
                filtered_df = await filter_existing_data(df)
                if filtered_df.empty:
                    logger.info(f"All data for trade_date {trade_date} already exists, skipping")
                    continue

                batch_dfs.append(filtered_df)

                # 每10个交易日的数据合并存储一次
                if len(batch_dfs) >= batch_size or i == total_dates - 1:
                    combined_df = pd.concat(batch_dfs, ignore_index=True)
                    
                    # 存储数据
                    @sync_to_async
                    def save_batch_data(df):
                        with transaction.atomic():
                            # 批量创建新记录
                            records = [
                                MarketTopList(**row)
                                for row in df.to_dict('records')
                            ]
                            try:
                                MarketTopList.objects.bulk_create(records, batch_size=1000, ignore_conflicts=True)
                                return len(records)
                            except Exception as e:
                                logger.warning(f"部分数据已存在，跳过重复数据: {str(e)}")
                                return 0
                    
                    count = await save_batch_data(combined_df)
                    count_all += count
                    batch_dfs = []  # 清空当前批次
                    logger.info(f"Saved batch of {count} records")

        except Exception as e:
            logger.error(f"Error updating market lhb list data: {str(e)}")
            raise

        # 更新数据更新记录
        await update_data_record(
            DATA_TYPES['MARKET_LHB_LIST'],
            end_date_obj
        )
        logger.info(f"Successfully updated {count_all} records for market lhb list data")
        return count_all
    async def update_market_lhb_detail(self, start_date=None, end_date=None, websocket=None):
        """
        更新龙虎榜每日机构交易明细数据
        :param start_date: 开始日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param end_date: 结束日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        # 解析日期
        try:
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近1年的数据
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=365)
        except Exception as e:
            logger.error(f"Error parsing dates: {str(e)}")
            raise

        # 获取所有交易日列表
        trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
            cal_date__gte=start_date_obj.strftime('%Y%m%d'),
            cal_date__lte=end_date_obj.strftime('%Y%m%d'),
            is_open=True
        ).order_by('cal_date').values_list('cal_date', flat=True))
        if not trade_dates:
            logger.warning("No trading days found in the specified date range")
            return 0

        total_dates = len(trade_dates)
        logger.info(f"Total trading days to process: {total_dates}")
        count_all = 0
        batch_size = 10  # 每10个交易日的数据合并存储一次
        batch_dfs = []  # 存储当前批次的数据
        
        # 初始化调用计数器和时间记录
        call_count = 0
        start_time = time.time()

        try:
            for i, trade_date in enumerate(trade_dates):
                # 发送进度更新
                if websocket:
                    message = f"正在更新龙虎榜机构交易明细数据 trade_date = {trade_date}"
                    try:
                        await websocket.send_progress(
                            current=i,
                            total=total_dates,
                            message=message
                        )
                    except Exception as e:
                        logger.error(f"WebSocket发送失败: {str(e)}")
                
                # 检查调用频率
                call_count += 1
                if call_count >= TUSHARE_API_LIMIT['market_lhb_top_list_max_in_minute']:
                    elapsed_time = time.time() - start_time
                    if elapsed_time < SECONDS_IN_MINUTE:  # 如果不足1分钟
                        await asyncio.sleep(SECONDS_IN_MINUTE - elapsed_time)
                    # 重置计数器和时间记录
                    call_count = 0
                    start_time = time.time()
                
                # 获取龙虎榜机构交易明细数据
                df = self.pro.top_inst(trade_date=trade_date)
                if df.empty:
                    logger.warning(f"No data found for trade_date: {trade_date}")
                    continue

                # 检查数据库中是否已存在相同 trade_date 和 ts_code 的数据
                @sync_to_async
                def filter_existing_data(df):
                    existing_records = MarketTopDetail.objects.filter(
                        trade_date__in=df['trade_date'].tolist(),
                        ts_code__in=df['ts_code'].tolist()
                    ).values_list('trade_date', 'ts_code')
                    existing_pairs = set((record[0], record[1]) for record in existing_records)
                    return df[~df.apply(lambda row: (row['trade_date'], row['ts_code']) in existing_pairs, axis=1)]
                
                filtered_df = await filter_existing_data(df)
                if filtered_df.empty:
                    logger.info(f"All data for trade_date {trade_date} already exists, skipping")
                    continue

                batch_dfs.append(filtered_df)

                # 每10个交易日的数据合并存储一次
                if len(batch_dfs) >= batch_size or i == total_dates - 1:
                    combined_df = pd.concat(batch_dfs, ignore_index=True)
                    
                    # 存储数据
                    @sync_to_async
                    def save_batch_data(df):
                        with transaction.atomic():
                            # 转换日期格式为YYYY-MM-DD
                             # 批量创建新记录
                            records = [
                                MarketTopDetail(**row)
                                for row in df.to_dict('records')
                            ]
                            
                            try:
                                MarketTopDetail.objects.bulk_create(records, batch_size=1000, ignore_conflicts=True)
                                return len(records)
                            except Exception as e:
                                logger.error(f"批量存储数据失败: {str(e)}")
                                return 0
                    
                    count = await save_batch_data(combined_df)
                    count_all += count
                    batch_dfs = []  # 清空当前批次
                    logger.info(f"Saved batch of {count} records")

        except Exception as e:
            logger.error(f"Error updating market lhb detail data: {str(e)}")
            raise

        # 更新数据更新记录
        await update_data_record(
            DATA_TYPES['MARKET_LHB_DETAIL'],
            end_date_obj
        )
        logger.info(f"Successfully updated {count_all} records for market lhb detail data")
        return count_all
        
    async def update_market_hm_detail(self, start_date=None, end_date=None, websocket=None):
        """
        更新游资交易明细数据
        :param start_date: 开始日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param end_date: 结束日期，格式：YYYYMMDD 或 YYYY-MM-DD
        :param websocket: WebSocket对象，可选
        :return: 更新的记录数
        """
        try:
            logger.info(f"开始更新游资交易明细数据，从 {start_date} 到 {end_date}")
            
            # 获取交易日历
            @sync_to_async
            def get_trading_dates(start_date, end_date):
                # 将日期转换为字符串格式
                start_str = datetime.strptime(start_date, '%Y%m%d').strftime('%Y%m%d')
                end_str = datetime.strptime(end_date, '%Y%m%d').strftime('%Y%m%d')
                
                # 查询交易日历
                trading_days = TradeDate.objects.filter(
                    cal_date__gte=start_str,
                    cal_date__lte=end_str,
                    is_open=True
                ).order_by('-cal_date').values_list('cal_date', flat=True)
                
                return list(trading_days)
            
            trading_days = await get_trading_dates(start_date, end_date)
            
            if not trading_days:
                logger.warning(f"在 {start_date} 到 {end_date} 期间没有找到交易日")
                return 0
                
            # 只取最后两个交易日
            last_two_days = trading_days[:2]
            logger.info(f"将更新以下交易日的游资交易明细数据: {last_two_days}")
            
            # 获取游资交易明细数据
            count_all = 0
            for trade_date in last_two_days:
                try:
                    df = self.pro.hm_detail(trade_date=trade_date)
                    if df.empty:
                        logger.warning(f"未获取到 {trade_date} 的游资交易明细数据")
                        continue
                    
                    # 保存数据到数据库
                    @sync_to_async
                    def save_hm_detail(df):
                        with transaction.atomic():
                            # 删除该交易日已有的数据
                            MarketHMDetailDaily.objects.filter(trade_date=trade_date).delete()
                            
                            # 准备要创建的记录
                            records = [
                                MarketHMDetailDaily(**row)
                                for row in df.to_dict('records')
                            ]
                            
                            # 批量创建新记录
                            MarketHMDetailDaily.objects.bulk_create(records, batch_size=1000)
                            return len(records)
                    
                    count = await save_hm_detail(df)
                    count_all += count
                    logger.info(f"成功保存 {count} 条 {trade_date} 的游资交易明细记录")
                    
                except Exception as e:
                    logger.error(f"获取或保存 {trade_date} 的游资交易明细数据时出错: {str(e)}")
                    continue
            
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['MARKET_HM_DETAIL'],
                datetime.strptime(end_date, '%Y%m%d').date()
            )
            
            logger.info(f"成功更新 {count_all} 条游资交易明细记录")
            return count_all
            
        except Exception as e:
            logger.error(f"更新游资交易明细数据时出错: {str(e)}")
            raise

        
    async def update_market_hm_info(self):
        """
        更新游资信息数据
        :return: 更新的记录数
        """
        try:
            logger.info("开始更新游资信息数据")
            
            # 获取游资名录数据
            df = self.pro.hm_list()
            if df.empty:
                logger.warning("未获取到游资名录数据")
                return 0
                
            # 处理orgs字段，拆分为多条记录
            records = []
            for _, row in df.iterrows():
                orgs = eval(row['orgs']) if isinstance(row['orgs'], str) else row['orgs']
                if not isinstance(orgs, list):
                    orgs = [orgs]
                
                # 每个org生成一条记录
                for org in orgs:
                    records.append({
                        'name': row['name'],
                        'desc': row['desc'],
                        'org': org
                    })
            
            # 清空原有数据并保存新数据
            @sync_to_async
            def clear_and_save(records):
                with transaction.atomic():
                    # 清空原有数据
                    MarketHMInfo.objects.all().delete()
                    
                    # 准备要创建的记录
                    to_create = []
                    for record in records:
                        hm_info = MarketHMInfo(
                            name=record['name'],
                            desc=record['desc'],
                            orgs=record['org']
                        )
                        to_create.append(hm_info)
                    
                    # 批量创建新记录
                    MarketHMInfo.objects.bulk_create(to_create, batch_size=1000)
                    
                    return len(to_create)
            
            # 执行数据库操作
            count = await clear_and_save(records)
            logger.info(f"成功更新 {count} 条游资信息记录")
            
            return count
            
        except Exception as e:
            logger.error(f"更新游资信息数据时出错: {str(e)}")
            raise
    async def update_market_hot(self, start_date, end_date, websocket=None):
        """
        更新市场热榜数据（同花顺和东方财富）
        :param start_date: 开始日期，datetime.date对象
        :param end_date: 结束日期，datetime.date对象
        :param websocket: WebSocket对象，可选
        """
        try:
             # 解析日期
            if start_date and end_date:
                if '-' in start_date:  # 格式为YYYY-MM-DD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    start_date_obj = dt.datetime.strptime(start_date, '%Y%m%d').date()
                
                if '-' in end_date:  # 格式为YYYY-MM-DD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y-%m-%d').date()
                else:  # 格式为YYYYMMDD
                    end_date_obj = dt.datetime.strptime(end_date, '%Y%m%d').date()
            else:
                # 如果没有指定日期范围，默认更新最近10个交易日
                end_date_obj = dt.date.today()
                start_date_obj = end_date_obj - dt.timedelta(days=20)  # 多取几天以确保包含10个交易日

            start_date_obj = max(start_date_obj, dt.date(2025, 1, 1))  # 确保开始日期不早于2019年1月1日

            # 获取交易日列表
            trade_dates = await sync_to_async(list)(TradeDate.objects.filter(
                cal_date__gte=start_date_obj.strftime('%Y%m%d'),
                cal_date__lte=end_date_obj.strftime('%Y%m%d'),
                is_open=True
            ).order_by('cal_date').values_list('cal_date', flat=True))

            if not trade_dates:
                logger.warning("No trading days found in the specified date range")
                return 0

            
            if not trade_dates:
                logger.warning(f"在 {start_date} 到 {end_date} 期间没有找到交易日")
                return 0

            # 遍历交易日
            for trade_date in trade_dates:
                print(f'更新 {trade_date} 的市场热榜数据')
                trade_date_str = trade_date.strftime('%Y%m%d') if isinstance(trade_date, dt.date) else trade_date
                
                # 获取同花顺热榜数据
                ths_hot_data_dfs = []
                for market_type in ['热股','行业板块','概念板块']:
                    ths_hot_df = self.pro.ths_hot(trade_date=trade_date_str, market=market_type)
                    ths_hot_data_dfs.append(ths_hot_df)
                ths_hot_data = pd.concat(ths_hot_data_dfs)


                @sync_to_async
                def save_ths_hot(df):
                    with transaction.atomic():
                        if not df.empty:
                        # 存储到MarketHotListTHS模型
                            for _, row in df.iterrows():
                                    MarketHotListTHS.objects.update_or_create(
                                        trade_date=trade_date_str,
                                        ts_code=row['ts_code'],
                                        defaults={
                                            'data_type': row['data_type'],
                                            'ts_name': row['ts_name'],
                                            'rank': row['rank'],
                                            'pct_change': row['pct_change'],
                                            'current_price': row['current_price'],
                                            'hot': row['hot'],
                                            'concept': row['concept'],
                                            'rank_time': row['rank_time'],
                                            'rank_reason': row['rank_reason']
                                        }
                                    )
                            return len(df)
                        else:
                            return 0
                @sync_to_async
                def save_ths_dc(df):
                    with transaction.atomic():
                        if not df.empty:
                            # 存储到MarketHotListDc模型
                            for _, row in df.iterrows():
                                MarketHotListDc.objects.update_or_create(
                                    trade_date=trade_date_str,
                                    ts_code=row['ts_code'],
                                    defaults={
                                        'data_type': row['data_type'],
                                        'ts_name': row['ts_name'],
                                        'rank': row['rank'],
                                        'pct_change': row['pct_change'],
                                        'current_price': row['current_price'],
                                        'hot': row['hot'],
                                        'concept': row['concept'],
                                        'rank_time': row['rank_time']
                                    }
                                )
                            return len(df)
                        else:
                            return 0
                # 保存数据
                count_ths = await save_ths_hot(ths_hot_data)
                # 获取东方财富热榜数据
                dc_hot_data = self.pro.dc_hot(trade_date=trade_date_str,market='A股市场')
                count_dc = await save_ths_dc(dc_hot_data)
                logger.info(f"成功更新 {count_ths+count_dc} 条热榜信息记录")
                
                # 更新进度
                if websocket:
                    await websocket.send_progress(
                        current=len(trade_dates),
                        total=len(trade_dates),
                        message=f"更新热榜数据: {trade_date_str}"
                    )
            # 更新数据更新记录
            await update_data_record(
                DATA_TYPES['MARKET_HOT_LIST'],
                datetime.strptime(end_date, '%Y%m%d').date()
            )

            logger.info(f"Successfully updated market hot data from {start_date} to {end_date}")
        except Exception as e:
            logger.error(f"Error updating market hot data: {str(e)}")
            raise